package com.adcc.ags.utility.mq.transfer;

import com.adcc.ags.utility.mq.configuration.MQConfiguration;
import com.adcc.ags.utility.mq.entity.Message;
import com.adcc.ags.utility.mq.entity.QMState;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.ibm.mq.MQException;
import com.ibm.mq.MQQueueManager;

import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * MQTransfer
 */
public abstract class MQTransfer {

    // MQConfiguration
    protected MQConfiguration configuration;

    // MQConnectionPool
    protected MQConnectionPool connectionPool;

    // 异步QM连接状态
    protected QMState asyncState = QMState.Disconnect;

    // 异步QM监控轮询时间
    protected int interval = 2000;

    // QMStateListener
    protected QMStateListener qmStateListener;

    // QueueMsgListener
    protected QueueMsgListener queueMsgListener;

    // TopicMsgListener
    protected TopicMsgListener topicMsgListener;

    // 队列异步接收队列
    protected List<AsyncReceiver> asyncQReceivers = Lists.newArrayListWithCapacity(1000);

    // 会话异步接收队列
    protected List<AsyncReceiver> asyncTReceivers = Lists.newArrayListWithCapacity(1000);

    /**
     * 构造函数
     */
    public MQTransfer(){

    }

    /**
     * 构造函数
     * @param configuration
     */
    public MQTransfer(MQConfiguration configuration){
        this.configuration = configuration;
    }

    /**
     * 构造函数
     * @param configuration
     * @param connectionPool
     */
    public MQTransfer(MQConfiguration configuration,MQConnectionPool connectionPool){
        this.configuration = configuration;
        this.connectionPool = connectionPool;
        this.connectionPool.initConnectionManager();
    }

    /**
     * 设置配置
     * @param configuration
     */
    public void setConfiguration(MQConfiguration configuration){
        this.configuration = configuration;
    }

    /**
     * 设置连接池
     * @param connectionPool
     */
    public void setConnectionPool(MQConnectionPool connectionPool){
        this.connectionPool = connectionPool;
        this.connectionPool.initConnectionManager();
    }

    /**
     * 设置监控轮询时间
     * @param interval
     */
    public void setInterval(int interval){
        this.interval = interval;
    }

    /**
     * 获取队列异步接收队列
     * */
    public List<AsyncReceiver> getAsyncQReceivers() {
        return asyncQReceivers;
    }

    /**
     * 获取会话异步接收队列
     * */
    public List<AsyncReceiver> getAsyncTReceivers() {
        return asyncTReceivers;
    }

    /**
     * 设置QM状态接收监听器
     * @param listener
     */
    public abstract void setQMStateListener(QMStateListener listener);

    /**
     * 设置队列消息接收监听器
     * @param listener
     */
    public abstract void setQueueMsgListener(QueueMsgListener listener);

    /**
     * 设置队列消息接收监听器
     * @param listener
     * @param queue
     */
    public abstract void setQueueMsgListener(QueueMsgListener listener, String...queue) throws Exception;

    /**
     * 设置会话接收监听器
     * @param listener
     */
    public abstract void setTopicMsgListener(TopicMsgListener listener);

    /**
     * 设置会话接收监听器
     * @param listener
     * @param topic
     */
    public abstract void setTopicMsgListener(TopicMsgListener listener, String...topic) throws Exception;

    /**
     * 是否连接QueuManager
     * @return
     * @throws Exception
     */
    public abstract boolean isConnectQM();

    /**
     * 从队列接收消息
     * @param queue
     * @return
     */
    public abstract Optional<Message> receiveQueue(String queue) throws Exception;

    /**
     * 从队列接收带头部的消息
     * @param queue
     * @param headKey
     * @return
     * @throws Exception
     */
    public abstract Optional<Message> receiveQueue(String queue,String headKey) throws Exception;

    /**
     * 从队列接收带头部的消息
     * @param queue
     * @param headKeys
     * @return
     * @throws Exception
     */
    public abstract Optional<Message> receiveQueue(String queue,Set<String> headKeys) throws Exception;

    /**
     * 向队列发送消息
     * @param queue
     * @param message
     */
    public abstract void sendQueue(String queue,byte[] message) throws Exception;

    /**
     * 向队列发送消息
     * @param queue
     * @param message
     * @param headKey
     * @param headValue
     */
    public abstract void sendQueue(String queue,byte[] message,String headKey,String headValue) throws Exception;

    /**
     * 向队列发送消息
     * @param queue
     * @param message
     * @param properties
     */
    public abstract void sendQueue(String queue,byte[] message,Map<String, String> properties) throws Exception;

    /**
     * 取得队列深度
     * @param queue
     * @throws Exception
     */
    public abstract int getDepth(String queue) throws Exception;

    /**
     * 向主题发送消息
     * @param topic
     * @param message
     * @param expiry 消息超时时间，单位0.1秒
     * @throws Exception
     */
    public abstract void sendTopic(String topic,byte[] message,int expiry) throws Exception;

    /**
     * 接收主题消息
     * @param topic
     * @return
     * @throws Exception
     */
    public abstract Optional<byte[]> receiveTopic(String topic) throws Exception;

    /**
     * 启动异步处理
     */
    public abstract void startAsync();

    /**
     * 关闭异步处理
     */
    public abstract void stopAsync();

    /**
     * 取得QueueManager
     * @return
     */
    protected abstract MQQueueManager getQueueManager() throws MQException;

    /**
     * 返还QueueManager
     * @param mqm
     * @throws Exception
     */
    protected abstract void returnQueueManager(MQQueueManager mqm) throws MQException;
}
